package com.datasync.batch;

import com.datasync.core.DataSourceFactory;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.database.JdbcCursorItemReader;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.transaction.PlatformTransactionManager;

import javax.sql.DataSource;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Map;

@Configuration
@EnableBatchProcessing
public class SyncJobConfig {

    @Autowired
    private JobRepository jobRepository;

    @Autowired
    private PlatformTransactionManager transactionManager;

    @Bean
    public Job dataSyncJob(Step fullSyncStep, Step incrementalSyncStep) {
        return new JobBuilder("dataSyncJob", jobRepository)
                .start(fullSyncStep)
                .next(incrementalSyncStep)
                .build();
    }

    @Bean
    public Step incrementalSyncStep(ItemReader<Map<String, Object>> incrementalReader,
                                   ItemProcessor<Map<String, Object>, Map<String, Object>> incrementalProcessor,
                                   ItemWriter<Map<String, Object>> jdbcWriter) {
        return new StepBuilder("incrementalSyncStep", jobRepository)
                .<Map<String, Object>, Map<String, Object>>chunk(100, transactionManager)
                .reader(incrementalReader)
                .processor(incrementalProcessor)
                .writer(jdbcWriter)
                .build();
    }

    @Bean
    public Step fullSyncStep(ItemReader<Map<String, Object>> reader,
                             ItemProcessor<Map<String, Object>, Map<String, Object>> processor,
                             ItemWriter<Map<String, Object>> writer) {
        return new StepBuilder("fullSyncStep", jobRepository)
                .<Map<String, Object>, Map<String, Object>>chunk(1000, transactionManager)
                .reader(reader)
                .processor(processor)
                .writer(writer)
                .build();
    }

    @Bean
    public ItemReader<Map<String, Object>> jdbcItemReader(DataSource sourceDataSource,
                                                           SyncJobProperties properties) {
        return new JdbcCursorItemReader<Map<String, Object>>() {{
            setDataSource(sourceDataSource);
            setSql("SELECT * FROM " + properties.getSourceTable());
            setRowMapper(new ColumnMapRowMapper());
        }};
    }

    @Bean
    public ItemProcessor<Map<String, Object>, Map<String, Object>> fieldMappingProcessor(
            SyncJobProperties properties) {
        return item -> {
            Map<String, String> fieldMappings = properties.getFieldMappings();
            fieldMappings.forEach((src, tgt) -> {
                if(item.containsKey(src)) {
                    item.put(tgt, item.remove(src));
                }
            });
            return item;
        };
    }

    @Bean
    public ItemWriter<Map<String, Object>> jdbcItemWriter(DataSource targetDataSource,
                                                          SyncJobProperties properties) {
        return new JdbcBatchItemWriter<Map<String, Object>>() {{
            setDataSource(targetDataSource);
            setSql("INSERT INTO " + properties.getTargetTable() + " (" +
                    String.join(",", properties.getFieldMappings().values()) + ") VALUES (" +
                    properties.getFieldMappings().values().stream()
                            .map(f -> ":" + f).reduce((a, b) -> a + "," + b).orElse("") + ")");
            setItemSqlParameterSourceProvider(map -> new MapSqlParameterSource(map));
        }};
    }

    private static class ColumnMapRowMapper implements RowMapper<Map<String, Object>> {
        @Override
        public Map<String, Object> mapRow(ResultSet rs, int rowNum) throws SQLException {
            return new LinkedCaseInsensitiveMap<>() {{
                ResultSetMetaData metaData = rs.getMetaData();
                for (int i = 1; i <= metaData.getColumnCount(); i++) {
                    put(metaData.getColumnName(i), rs.getObject(i));
                }
            }};
        }
    }
}